package com.intct.dws;

import com.intct.common.FlinkSqlWithUtil;
import com.intct.func.CurrentTime;
import com.intct.func.RowKeyFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

/**
 * @author gufg
 * @since 2025-10-25 16:36
 */
public class DwsOrderSQL {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        env.setParallelism(1);
         env.enableCheckpointing(5000L);//每五秒一个快照
        /*{
          "id": 72,
          "order_no": "176120846699712287857",
          "order_type": 0,
          "state": 1,
          "create_time": "2025-10-23 16:34:27",
          "update_time": "2025-10-23 16:56:11",
          "pay_amount": 6.0
        }*/

        //kafka这里不支持主键
        tenv.executeSql(
                "CREATE TABLE order_kafka (\n" +
                        "    id INT,\n" +
                        "    order_no STRING,\n" +
                        "    order_type  INT,\n" +//订单类型
                        "    state INT,\n" +//下单
                        "     create_time timestamp(3),\n" +
                        "     update_time timestamp(3),\n" +
                        "    pay_amount double,\n" +
                        "    WATERMARK FOR create_time AS create_time - INTERVAL '5' SECOND\n" +
                        ")" + FlinkSqlWithUtil.getKafkaSourceWith("order_Test","order_kafka", ""));
        //tenv.executeSql("select * from order_kafka").print();

        // 注册一个返回当前时间的函数
        tenv.createTemporarySystemFunction("CU", CurrentTime.class);
        tenv.createTemporarySystemFunction("rowkeyFunc", RowKeyFunction.class);

        // 1. 先注册过滤后的视图
        tenv.executeSql(
                "CREATE VIEW filtered_orders AS " +
                        "SELECT * FROM order_kafka WHERE create_time <= CU()"
        );

        tenv.executeSql("CREATE TEMPORARY VIEW END_VIEW AS " +
                "SELECT\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                // 订单总数：order_type=0或1（不管state，只要类型符合）
                "    COUNT(CASE WHEN order_type IN (0, 1) THEN order_no ELSE NULL END) AS count_order,\n" +
                // 总额：order_type=0或1 且 state=5 或6
                "    SUM(CASE WHEN order_type IN (0, 1) AND (state = 5 or state = 6 ) THEN pay_amount ELSE 0 END) AS sum_money,\n" +
                // 应付总额：order_type=0或1 且 state=5
                "    SUM(CASE WHEN order_type IN (0, 1) AND state = 5 THEN pay_amount ELSE 0 END) AS maybe_money\n" +
                "FROM TABLE(\n" +
                "    CUMULATE(\n" +//累计窗口
                "        TABLE filtered_orders ,\n" +  // 表名
                // "        TABLE ( select * from order_kafka WHERE create_time <= CU()),\n" +  // 表名
                //这里的where不能直接加，要不写子查询要不写视图  视图提前过滤
                // "        WHERE create_time <= CU(), "+ // 只统计当前时间之前的订单
                "        DESCRIPTOR(create_time),\n" +//watermark
                "        INTERVAL '1' SECOND,\n" +//顺序不能换   步长  窗口更新频率
                "        INTERVAL '1' DAY\n" +//窗口大小
                "    )\n" +
                ") GROUP BY window_start, window_end"
        );
        // tenv.executeSql("select * from  END_VIEW").print();
        //create 'travel:order_st', 'cf', SPLITS => ['100|', '200|', '300|', '400|']
        tenv.executeSql("CREATE TABLE  order_stats (\n" +
                "    id STRING,\n" +
                "    cf ROW< " +
                "    window_start STRING,\n" +
                "    window_end  STRING,\n" +//订单类型
                "    count_order STRING,\n" +//下单
                "    sum_money STRING,\n" +
                "    maybe_money STRING >"+
                ")"+
                FlinkSqlWithUtil.getHBaseSinkWith("travel","order_st"));

        // 将 END_VIEW 数据写入 HBase 的 order_stats 表
        tenv.executeSql("INSERT INTO order_stats\n" +
                "SELECT\n" +
                "     rowkeyFunc(CONCAT(CAST(window_start AS STRING), '_', CAST(window_end AS STRING)),5) AS rowkey,\n" +
                "    ROW(\n" +
                "        CAST(window_start AS STRING), \n" +
                "        CAST(window_end AS STRING),    \n" +
                "        CAST(count_order AS STRING),  \n" +
                "        CAST(sum_money AS STRING),    \n" +
                "        CAST(maybe_money AS STRING)  \n" +
                "    ) AS cf\n" +
                "FROM END_VIEW"  // 从视图读取数据
        );
    }
}
